package com.atguigu.edu.realtime.app.dwd.db;

import com.atguigu.edu.realtime.utils.MyKafkaUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @ClassName DwdUserLoginOperation
 * @Description TODO
 * @Author$ 邢家俊
 * @Date 2023-5-5 22:05
 * @Version 1.0
 * 用户登录事务事实表
 **/
public class DwdUserLoginOperation {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        String groupId="topic_user_Login_group";
        tableEnv.executeSql(MyKafkaUtil.getTopicDbDDL(groupId));
//  "data['id'],data['login_name'],data['real_name'],data['phone_num'],data['email'],data['user_level'],data['birthday'],data['create_time'] +
//                +
        Table userLogin = tableEnv.sqlQuery("select " +
                " data['id']," +
                " data['login_name']," +
                " data['real_name']," +
                " data['phone_num']," +
                " data['email']," +
                " data['user_level']," +
                " data['birthday']," +
                " date_format(data['create_time'], 'yyyy-MM-dd') date_id,"+
                " type ,ts "+
                " from `topic_db` where `table` = 'user_info'");

        tableEnv.createTemporaryView("user_login", userLogin);

        tableEnv.executeSql(" create table dwd_user_login_Operation( " +
                " id string ,"+
                " `login_name` string,"+
                " `real_name` string, " +
                " `phone_num` string ,"+
                " `email` string ,"+
                " `user_level` string ,"+
                " `birthday` string ,"+
                " date_id string ,"+
                " type string ,"+
                " ts string, " +
                " primary key(id) not enforced)"
                +MyKafkaUtil.getUpsertKafkaDDL("dwd_user_login_Operation"));
        tableEnv.executeSql("insert into dwd_user_login_Operation select * from user_login");
    }
}
